/**
 * Created by IntelliJ IDEA.
 * User: jiyufei
 * Date: 2018/10/11
 * Time: 13:55
 **/
package com.jiyufei.linux.main;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import com.jiyufei.linux.bean.YumLog;
import com.jiyufei.linux.bolt.*;
import storm.kafka.*;

import java.util.Arrays;
import java.util.Scanner;

public class HlasLinuxLog {
    
    
    private static final String KAFKA_SPOUT_ID = "kafka-spout";
    private static final String SWITCH_BOLT_ID = "switch-bolt";
    private static final String SECURE_SPLIT_BOLT_ID = "secure-split-bolt";
    private static final String MESSAGE_SPLIT_BOLT_ID = "message-split-bolt";
    private static final String CRON_SPLIT_BOLT_ID = "cron-split-bolt";
    private static final String YUM_SPLIT_BOLT_ID = "yum-split-bolt";
    
    
    private static final String MESSAGE_ES_BOLT_ID = "message-es-bolt";
    private static final String SECURE_ES_BOLT_ID = "secure-es-bolt";
    private static final String CRON_ES_BOLT_ID = "cron-es-bolt";
    private static final String YUM_ES_BOLT_ID = "yum-es-bolt";
    
    
    private static final String TOPOLOGY_NAME = "HlasYunYingLinuxLog";
    
    
    
    public static void main(String args[]){
        
        SwitchBolt switchBolt = new SwitchBolt();
        SecureSplitBolt secureSplitBolt = new SecureSplitBolt();
        OtherBolt otherBolt = new OtherBolt();
        MessageSplitBolt messageSplitBolt = new MessageSplitBolt();
        CronSplitBolt cronSplitBolt = new CronSplitBolt();
        YumSplitBolt yumSplitBolt = new YumSplitBolt();
        
        SecureElasticSearchBolt secureElasticSearchBolt = new SecureElasticSearchBolt();
        MessageElasticSearchBolt messageElasticSearchBolt = new MessageElasticSearchBolt();
        CronElasticSearchBolt cronElasticSearchBolt = new CronElasticSearchBolt();
        YumElasticSearchBolt yumElasticSearchBolt = new YumElasticSearchBolt();
        
        String zkRoot = "/kafka-spout";
        String id = "storm-consumer-yunying-linux";
        
        BrokerHosts brokerHosts = new ZkHosts("192.168.32.33:2181,192.168.32.34:2181,192.168.32.35:2181,192.168.32.36:2181","/kafka/brokers");
        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts,"yunying-linux-log",zkRoot,id);
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        spoutConfig.zkServers = Arrays.asList(new String[]{"192.168.32.33,192.168.32.34,192.168.32.35,192.168.32.36"});
        spoutConfig.zkPort = 2181;
        spoutConfig.zkRoot = zkRoot;
        spoutConfig.forceFromStart = false;
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
        
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout(KAFKA_SPOUT_ID,kafkaSpout,2);
        topologyBuilder.setBolt(SWITCH_BOLT_ID,switchBolt,3).shuffleGrouping(KAFKA_SPOUT_ID);
        
        topologyBuilder.setBolt(SECURE_SPLIT_BOLT_ID,secureSplitBolt,1).globalGrouping(SWITCH_BOLT_ID,"secure-stream");
        topologyBuilder.setBolt(SECURE_ES_BOLT_ID,secureElasticSearchBolt,1).globalGrouping(SECURE_SPLIT_BOLT_ID,"secure-stream");
        
        topologyBuilder.setBolt(MESSAGE_SPLIT_BOLT_ID,messageSplitBolt,1).globalGrouping(SWITCH_BOLT_ID,"message-stream");
        topologyBuilder.setBolt(MESSAGE_ES_BOLT_ID,messageElasticSearchBolt,1).globalGrouping(MESSAGE_SPLIT_BOLT_ID,"message-stream");
    
        topologyBuilder.setBolt(CRON_SPLIT_BOLT_ID,cronSplitBolt,1).globalGrouping(SWITCH_BOLT_ID,"cron-stream");
        topologyBuilder.setBolt(CRON_ES_BOLT_ID,cronElasticSearchBolt,1).globalGrouping(CRON_SPLIT_BOLT_ID,"cron-stream");
    
        topologyBuilder.setBolt(YUM_SPLIT_BOLT_ID,yumSplitBolt,1).globalGrouping(SWITCH_BOLT_ID,"yum-stream");
        topologyBuilder.setBolt(YUM_ES_BOLT_ID,yumElasticSearchBolt,1).globalGrouping(YUM_SPLIT_BOLT_ID,"yum-stream");
        
        
        topologyBuilder.setBolt("other-bolt",otherBolt,1).globalGrouping(SWITCH_BOLT_ID,"other-stream");
        
        
        Config config = new Config();
        if(args != null && args.length > 0){
            config.put(config.NIMBUS_HOST,args[0]);
            config.setNumWorkers(3);
            try {
                StormSubmitter.submitTopology(TOPOLOGY_NAME,config,topologyBuilder.createTopology());
                System.out.println("[HlasLinuxLog.main]业务拓扑" + TOPOLOGY_NAME + "提交成功！\r\n");
            }catch (AlreadyAliveException e){
                System.out.println("[HlasLinuxLog.main]AlreadyAliveException 业务拓扑" + TOPOLOGY_NAME + "已存在！\r\n" + e.getMessage());
            }catch (InvalidTopologyException e){
                System.out.println("[HlasLinuxLog.main]AlreadyAliveException 提交业务拓扑" + TOPOLOGY_NAME + "失败！\r\n" + e.getMessage());
            }
        }else{
            System.out.println("本地运行模式");
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(TOPOLOGY_NAME, config, topologyBuilder.createTopology());
            //Utils.sleep(120000);
            Scanner sc = new Scanner(System.in);
            while(true){
                if(sc.next().compareTo("kill") == 0){
                    cluster.killTopology(TOPOLOGY_NAME);
                    cluster.shutdown();
                    sc.close();
                }
            }
        }
    }
    
}
